package com.bleeth.flow.step.output;

import cn.hutool.core.thread.ThreadUtil;
import cn.hutool.http.ContentType;
import cn.hutool.http.server.SimpleServer;
import cn.hutool.json.JSONObject;
import com.bleeth.flow.core.common.APlugin;
import com.bleeth.flow.core.common.PluginAnnotation;
import com.bleeth.flow.core.common.PluginTypeEnum;
import com.jd.platform.async.worker.WorkResult;
import com.jd.platform.async.wrapper.WorkerWrapper;
import lombok.Data;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;

import java.io.Serializable;
import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

/**
 * @author ：Bleeth
 * @date ：2021-08-06
 * @description：一个将数据输出到Http接口的组件(动态接口)
 */
@Data
@PluginAnnotation(name = "http写",
        type = PluginTypeEnum.OUTPUT,
        description = "",
        id = "HttpWriterPlugin")
public class HttpWriterPlugin extends APlugin implements Serializable {


    private static final long serialVersionUID = -2174529745200727398L;


    private String host;

    private Integer port;

    private String urlPath;


    @Override
    public void result(boolean success, Dataset<Row> param, WorkResult<Dataset<Row>> workResult) {

    }

    @Override
    public Dataset<Row> action(Dataset<Row> param, Map<String, WorkerWrapper> allWrappers) {
        super.action(param, allWrappers);

        new SimpleServer(host, port)
                .addAction(urlPath, (request, response) -> {

                    String fromPluginName = fromList.get(0);
                    WorkerWrapper fromWrapper = allWrappers.get(fromPluginName);
                    Dataset<Row> ds = (Dataset<Row>) fromWrapper.getWorkResult().getResult();

                    List<Row> rowList = ds.collectAsList();
                    List<JSONObject> objectList = new ArrayList<>();
                    StructField[] fieldArray = null;
                    for (int i = 0; i < rowList.size(); i++) {
                        Row row = rowList.get(i);
                        if (i == 0) {
                            StructType schema = row.schema();
                            fieldArray = schema.fields();
                        }

                        JSONObject object = new JSONObject();
                        objectList.add(object);
                        for (int fieldIndex = 0; fieldIndex < fieldArray.length; fieldIndex++) {
                            StructField structField = fieldArray[fieldIndex];
                            object.set(structField.name(), row.get(fieldIndex));
                        }
                    }

                    //处理请求中...
                    response.sendOk();
                    response.setCharset(Charset.forName("UTF-8"));
                    JSONObject object = new JSONObject();
                    object.set("code", "200");
                    object.set("msg", "请求成功");
                    object.set("data", objectList);

                    response.setHeader("content-type","application/json; charset=utf-8");
                    response.setCharset(Charset.forName("utf8"));
                    response.write(object.toString(), ContentType.JSON.toString());
                }).start();

        ThreadUtil.sleep(10000 * 1000);
        return null;
    }


}
